package cn.com.demo.kafka.consumer;


import cn.com.demo.kafka.Config;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

import static org.apache.kafka.clients.consumer.ConsumerConfig.*;

/**
 * 消费数据
 */
public class ConsumerAutoCommitDemo {


    public static void main(String args[]) {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(conf());
        consumer.subscribe(Arrays.asList(Config.TOPIC));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                int recordCount = records.count();
                System.out.println("Received " + recordCount + " messages.");
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("Received message: " + record.value());
                    // 延迟消费消息
                    Thread.sleep(Config.CONSUMER_DEAL_MSG_INTERVAL);
                    //  手动确认消息
                     consumer.commitSync(Collections.singletonMap(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)));
                }
                // 延迟获取消息
                Thread.sleep(Config.CONSUMER_POLL_MSG_INTERVAL);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }


    }

    private static Properties conf() {
        Properties conf = new Properties();
        conf.setProperty(BOOTSTRAP_SERVERS_CONFIG, Config.BOOTSTRAP_SERVERS_CONFIG);
        conf.setProperty(GROUP_ID_CONFIG, Config.GROUP_ID_CONFIG);
        //conf.setProperty(AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        conf.setProperty(AUTO_OFFSET_RESET_CONFIG, "earliest");
        conf.setProperty(ENABLE_AUTO_COMMIT_CONFIG, "false"); // 禁用自动提交
        conf.setProperty(KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        conf.setProperty(VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        return conf;
    }
}